package com.leyantech.ecp.agenx.session.manager.consumer;

import static com.leyantech.ecp.agenx.common.utils.SplitterUtils.DOT_SPLITTER;

import com.leyantech.chaos.kafka.KafkaConsumerStringProcessor;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.List;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-07.
 */
@Slf4j
@Singleton
public class AssistantEventConsumer implements KafkaConsumerStringProcessor {

  private final List<ActionHandler> actionHandlers;

  @Inject
  public AssistantEventConsumer(List<ActionHandler> actionHandlers) {
    this.actionHandlers = actionHandlers;
  }

  @Override
  public void accept(ConsumerRecord<String, String> record) {
    final String message = record.value();

    final JsonObject clientMessage = new JsonObject(message);

    final JsonObject header = clientMessage.getJsonObject("header");
    final String fn = header.getString("fn");

    for (ActionHandler actionHandler : actionHandlers) {
      if (actionHandler.actionType().type.equals(fn)) {

        final List<String> storeIdAndAssistant = DOT_SPLITTER.splitToList(record.key());

        header.put("sid", storeIdAndAssistant.get(0))
            .put("aid", storeIdAndAssistant.get(1));

        actionHandler.handle(clientMessage);
        return;
      }
    }

  }

}
